package cn.doitedu.kafka.demos;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.RangeAssignor;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

public class ConsumerGroupReblance测试3 {

    public static void main(String[] args) {

        Properties props = new Properties();
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "doit01:9092,doit02:9092,doit03:9092");
        props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        // 如果消费者从__consumer_offset中获取不到之前记录的消费偏移量，则通过该参数来决定从什么位置拉取数据
        props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");  //  earliest|latest|none

        // 客户端id
        props.setProperty(ConsumerConfig.CLIENT_ID_CONFIG, "x03");

        // 消费者组ID
        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "g001");

        // 指定消费者再均衡分区分配策略为Range策略
        props.setProperty(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, RangeAssignor.class.getName());

        // 指定消费者再均衡分区分配策略为round robin（轮询）策略
        //props.setProperty(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, RoundRobinAssignor.class.getName());



        // 构造消费者实例
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);


        // 订阅主题
        consumer.subscribe(Arrays.asList("tpc-3","tpc-10"));

        //
        while(true){
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
            System.out.println(consumer.assignment());
        }


    }


}
